package com.simple.datax.loader.impl;

import java.util.List;

import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.api.Loader;
import com.simple.datax.api.Receiver;
import com.simple.datax.conn.BaseConnector;
import com.simple.datax.conn.ConnectorPool;
import com.simple.datax.conn.impl.ConnectorPoolManager;
import com.simple.datax.core.model.connpara.CommonConnPara;
import com.simple.datax.service.DBLoaderService;

public class DBScheduledLoaderImpl implements Loader {

	private DBLoaderService dbloader = null;
	private Receiver receiver = null;
	private ThreadPoolTaskExecutor taskExecutor = null;
	private int threshold;
	private ConnectorPool connectorPool = ConnectorPoolManager.getInstance().getConnectorPool(ConnectorPoolManager.CONN_RECV, null, null);
	
	private static final Logger logger = LoggerFactory
			.getLogger(DBScheduledLoaderImpl.class);
	
	@Async
	@Override
	public void run() {
		logger.debug("========" + "Load.run()");
		long start = System.currentTimeMillis();
		List<CommonConnPara> list = loadAllConnectorInfo();
		long end = System.currentTimeMillis();
		logger.debug("======== time:" + (end-start));
		if (list == null) {
			logger.debug("======== list is null.");
			return;
		}
		logger.debug("========" + list.size());
		BaseConnector connector = null;
		for(CommonConnPara para:list) {
			// get connector from pool
			try {
				connector = connectorPool.borrowObject(para);
				
				// got connector then continue
				if (connector != null) {
					taskExecutor.execute(receiver.process(connector, para, threshold));
					 // Spring controls creating Receiver object
				} else {
					logger.debug("No available Connector, skip this time.");
				}
			} catch (Exception e) {
				logger.error("Reciver get message error.", e);
			}
		}
	}
	
	private List<CommonConnPara> loadAllConnectorInfo() {
		logger.debug("loadAllConnectorInfo()");
		return dbloader.getRecvConnParas();
//		List<CommonConnPara> ret = new ArrayList<CommonConnPara>(1000);
//		CommonConnPara para = null;
//		
////		for (int i=0; i<100; i++) {
//			for (int j=1; j<2; j++) {
//				para = new CommonConnPara();
//				ret.add(para);
////				para.setTransportType(FTPConnImpl.TYPE);
////				IP$PORT$USERNAME$PASSWORD$WORKING_PATH$NEED_CLIENT_CERT$NEED_SERVER_CERT$CLIENT_KEYSTORE_PATH$SERVER_KEYSTORE_PATH$CLIENT_KEYSTORE_PASSWORD$SERVER_KEYSTORE_PASSWORD
////				para.setTransportPara("10.2.3.52$21$jy$jy$"+"/"+j); //FTP
//				
////				para.setTransportType(MSMQConnImpl.TYPE);
////				para.setTransportPara("DIRECT=TCP:10.2.3.52\\private$\\Test");// MSMQ
//				
////				para.setTransportType(IBMMQConnImpl.TYPE);
////				IP$PORT$USERNAME$PASSWORD$CCID$QUEUE_MANAGER_NAME$CHANNEL$QUEUE_NAME
////				para.setTransportPara("10.2.3.52$1500$mqadmin$Art515940$1381$RemoteSend$JAVA.CHANNEL$SendQueue"); //IBMMQ
//
//				para.setTransportType(ActiveMQConnImpl.TYPE);
////				TCP://IP:PORT$USERNAME$PASSWORD$QUEUE_NAME
//				para.setTransportPara("tcp://JoeJinWin7:61616$"+ActiveMQConnection.DEFAULT_USER+"$"+ActiveMQConnection.DEFAULT_PASSWORD+"$QUEUE_"+j); //ActiveMQ
//				
////				para.setTransportType(SFTPConnImpl.TYPE);
////				IP$PORT$USERNAME$PASSWORD$WORKING_PATH$
////				para.setTransportPara("111.13.47.157$2226$qicai$qicai$ $");
//				
////				para.setTransportType(HornetqConnImpl.TYPE);
////				IP$PORT$QUEUE_NAME
////				para.setTransportPara("localhost$5445$DLQ");
//				
////				para.setTransportType(LocalFileConnImpl.TYPE);
////				DIRECTORY
////				para.setTransportPara("C://");
//			}
////		}
//		return ret;
	}

	public Receiver getReceiver() {
		return receiver;
	}

	public void setReceiver(Receiver receiver) {
		this.receiver = receiver;
	}

	public ThreadPoolTaskExecutor getTaskExecutor() {
		return taskExecutor;
	}

	public void setTaskExecutor(ThreadPoolTaskExecutor taskExecutor) {
		this.taskExecutor = taskExecutor;
	}

	public DBLoaderService getDbloader() {
		return dbloader;
	}

	public void setDbloader(DBLoaderService dbloader) {
		this.dbloader = dbloader;
	}

	public int getThreshold() {
		return threshold;
	}

	public void setThreshold(int threshold) {
		this.threshold = threshold;
	}

}
